package com.azunitech.intercpt;

import java.net.URISyntaxException;

import org.apache.log4j.Logger;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.junit.Before;
import org.junit.Test;

public class MQTTTester {
	private static Logger logger = Logger.getLogger(MQTTTester.class);
	MQTT mqtt = null;
	FutureConnection connection = null;	
	private String sDestination = "foo";
	private String sAddress = "tcp://localhost:1883";
	private String sMessage = "mmmmmmm";
	private String sUserName = "admin";
	private String sPassword = "admin";
	
	@Before public void before() throws URISyntaxException{
		this.connect();
	}
	
	
	@Test public void test() throws Exception{
		MQTT mqtt = new MQTT();
		mqtt.setHost("localhost", 1883);
		mqtt.setHost("tcp://localhost:1883");
		BlockingConnection connection = mqtt.blockingConnection();
		connection.connect();	
		connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false);
		Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
		byte[] qoses = connection.subscribe(topics);
	}


	
	@Test public void test1() throws Exception{
		MQTT mqtt = new MQTT();
		//mqtt.setHost("localhost", 1883);
		mqtt.setHost("tcp://localhost:1883");

		FutureConnection connection = mqtt.futureConnection();
		Future<Void> f1 = connection.connect();
		f1.await();

		Future<byte[]> f2 = connection.subscribe(new Topic[]{new Topic( "foo", QoS.AT_LEAST_ONCE)});
		byte[] qoses = f2.await();

		// We can start future receive..
		Future<Message> receive = connection.receive();

		// send the message..
		Future<Void> f3 = connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false);

		// Then the receive will get the message.
		Message message = receive.await();
		message.ack();
		byte[] payload = message.getPayload();
		String messagePayLoad = new String(payload);
		System.out.println(messagePayLoad);
		Future<Void> f4  = connection.disconnect();
		f4.await();
	}

		
		@Test public void test2() throws InterruptedException{
			this.send();
			Thread.sleep(10000);
		}
		
		// callback used for Future
			<T> Callback<T> onui(final Callback<T> original) {
				return new Callback<T>() {
					public void onSuccess(final T value) {
						Runnable r = new Runnable(){
							public void run() {
								original.onSuccess(value);
							}};
						r.run();	
							
					}
					public void onFailure(final Throwable error) {
						Runnable r = new Runnable(){
							public void run() {
								original.onFailure(error);
							}};
						r.run();	
					}
				};
			}			
		private void send()
		{
			if(connection != null)
			{
				// automatically connect if no longer connected
				if(!connection.isConnected())
				{
					connect();
				}
				
				Topic[] topics = {new Topic(sDestination, QoS.AT_LEAST_ONCE)};
				connection.subscribe(topics).then(onui(new Callback<byte[]>() {
					public void onSuccess(byte[] subscription) {
						logger.info("Destination: " + sDestination);
						logger.info("Message: " + sMessage);
						
						// publish message
						connection.publish(sDestination, sMessage.getBytes(), QoS.AT_LEAST_ONCE, false);
						logger.info("Message sent");
						
						// receive message
						connection.receive().then(onui(new Callback<Message>() {
							public void onSuccess(Message message) {
								String receivedMesageTopic = message.getTopic();
								byte[] payload = message.getPayload();
								String messagePayLoad = new String(payload);
								message.ack();
								connection.unsubscribe(new String[]{sDestination});
								logger.info(receivedMesageTopic + ":" + messagePayLoad);
							}
							
							public void onFailure(Throwable e) {
								logger.info("Exception receiving message: " + e);
							}
						}));
						
					}
					
					public void onFailure(Throwable e) {
						logger.info("Exception sending message: " + e);
					}
				}));
			}
			else
			{
			}
		}
	
		private void disconnect()
		{
			try
			{
				if(connection != null && connection.isConnected())
				{
					connection.disconnect().then(onui(new Callback<Void>(){
						public void onSuccess(Void value) {
							logger.info("Disconnected");
						}
						public void onFailure(Throwable e) {
							logger.info("Problem disconnecting");
							logger.info("Exception disconnecting from " + sAddress + " - " + e);
						}
					}));
				}
				else
				{
					logger.info("Not Connected");
				}
			}
			catch(Exception e)
			{
				logger.info("Exception " + e);
			}
		}
		
		private void connect()
		{
			mqtt = new MQTT();
			mqtt.setClientId("android-mqtt-example");
	
			try
			{
				mqtt.setHost(sAddress);
				logger.info("Address set: " + sAddress);
			}
			catch(URISyntaxException urise)
			{
				logger.info("URISyntaxException connecting to " + sAddress + " - " + urise);
			}
			
			if(sUserName != null && !sUserName.equals(""))
			{
				mqtt.setUserName(sUserName);
				logger.info("UserName set: [" + sUserName + "]");
			}
			
			if(sPassword != null && !sPassword.equals(""))
			{
				mqtt.setPassword(sPassword);
				logger.info("Password set: [" + sPassword + "]");
			}
			
			connection = mqtt.futureConnection();
			connection.connect().then(onui(new Callback<Void>(){
				public void onSuccess(Void value) {
					logger.info("Connected");
				}
				public void onFailure(Throwable e) {
					logger.info("Problem connecting to host");
					logger.info("Exception connecting to " + sAddress + " - " + e);
				}
			}));
		}
}
